package com.mindata.blockchain.socket.raft.client;

import cn.hutool.core.collection.CollUtil;
import com.mindata.blockchain.ApplicationContextProvider;
import com.mindata.blockchain.block.Block;
import com.mindata.blockchain.block.Instruction;
import com.mindata.blockchain.block.check.CheckerManager;
import com.mindata.blockchain.common.ConsensusType;
import com.mindata.blockchain.common.ResultPack;
import com.mindata.blockchain.core.event.AddBlockEvent;
import com.mindata.blockchain.core.event.AddInstructionEvent;
import com.mindata.blockchain.core.manager.DbBlockManager;
import com.mindata.blockchain.socket.body.RpcBlockBody;
import com.mindata.blockchain.socket.body.RpcCheckBlockBody;
import com.mindata.blockchain.socket.body.RpcCheckBlockResult;
import com.mindata.blockchain.socket.client.PacketSender;
import com.mindata.blockchain.socket.packet.BlockPacket;
import com.mindata.blockchain.socket.packet.PacketBuilder;
import com.mindata.blockchain.socket.packet.PacketType;
import com.mindata.blockchain.socket.raft.server.RaftServerServiceContainer;
import com.mindata.blockchain.socket.raft.server.ServerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.tio.utils.json.Json;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


@Service
@Slf4j
public class RaftServiceImpl implements RaftService {
    @Resource
    private PacketSender packetSender;

    ExecutorService readThreadPool = Executors.newFixedThreadPool(3);
    ExecutorService writeThreadPool = Executors.newFixedThreadPool(3);

    @Override
    public void set(String key, Block block) throws Exception {
        ResultPack<String> resultPack = checkBlock(block);
        if (resultPack.isFailed()) {
            throw new Exception("区块校验失败:" + resultPack.comment());
        }
        String value = Json.toJson(block);
        ServerMessage.SetRequest setRequest = ServerMessage.SetRequest.newBuilder()
                .setKey(key).setValue(value).build();
        //提交写请求
        writeThreadPool.submit(new WriteTask(key, setRequest, readThreadPool, packetSender));
    }

    /**
     * @Author:wenhui
     * @Description:block校验
     * @Date:Create in 2018-7-30 15:37
     **/
    private ResultPack<String> checkBlock(Block block) {
        CheckerManager checkerManager = ApplicationContextProvider.getBean(CheckerManager.class);
        RpcCheckBlockBody rpcCheckBlockBody = checkerManager.check(block);
        if (rpcCheckBlockBody.getCode() == RpcCheckBlockResult.EXITS_INSTRUCTION.getCode()) {
            //如果指令中存在已经被打包的，找出未打包指令后本地再添加，重新再次打包
            List<Instruction> appendingInstructions = ApplicationContextProvider.getBean(DbBlockManager.class)
                    .checkAndGetpendingInstructions(block.getBlockBody().getInstructions());

            if (CollUtil.isNotEmpty(appendingInstructions)) {
                ApplicationContextProvider.publishEvent(new AddInstructionEvent(appendingInstructions));
            }
        }
        if (rpcCheckBlockBody.getCode() == RpcCheckBlockResult.SUCCESS.getCode()) {
            return ResultPack.succeed();
        } else {
            return ResultPack.failed(rpcCheckBlockBody.getMessage());
        }
    }

    /**
     * @Author:wenhui
     * @Description:
     * @Date:Create in 2018-7-30 16:23
     **/
    public class WriteTask implements Runnable {
        private String key;
        private ServerMessage.SetRequest setRequest;
        private ExecutorService readThreadPool;
        private PacketSender packetSender;

        public WriteTask(String key, ServerMessage.SetRequest setRequest, ExecutorService readThreadPool, PacketSender packetSender) {
            this.key = key;
            this.setRequest = setRequest;
            this.readThreadPool = readThreadPool;
            this.packetSender = packetSender;
        }

        @Override
        public void run() {
            ServerMessage.SetResponse setResponse = RaftServerServiceContainer.getInstance().set(setRequest);
            if (setResponse != null && setResponse.getSuccess()) {
                //raftserver端节点响应成功，即可轮询读取文件
                readThreadPool.submit(new ReadTask(packetSender, key));
                return;
            } else {
                log.error("raft set error key:{} value:{}", setRequest.getKey(), setRequest.getValue());
            }
        }
    }

    public class ReadTask implements Runnable {
        private String key;
        private PacketSender packetSender;

        public ReadTask(PacketSender packetSender, String key) {
            this.key = key;
            this.packetSender = packetSender;
        }

        @Override
        public void run() {
            ServerMessage.GetRequest getRequest = ServerMessage.GetRequest.newBuilder()
                    .setKey(key).build();
            while (true) {
                ServerMessage.GetResponse getResponse = RaftServerServiceContainer.getInstance().get(getRequest);
                try {
                    if (getResponse != null) {
                        Block block = Json.toBean(getResponse.getValue(), Block.class);
                        if (block == null) {
                            log.debug("get response block is null value:{}", getResponse.getValue());
                            continue;
                        }
                        //leader 生成block后也需要校验区块合法性
                        ResultPack<String> resultPack = checkBlock(block);
                        if (resultPack.isFailed()) {
                            log.error("区块校验失败:{}", resultPack.comment());
                            return;
                        }
                        //TODO add之前做共识处理
                        ApplicationContextProvider.publishEvent(new AddBlockEvent(ConsensusType.RAFT, block));
                        //全md_chain节点发送基于raft共识一致性消息
                        BlockPacket blockPacket = new PacketBuilder<>().setType(PacketType.COMMIT_RAFT_BLOCK_REQUEST.getKey()).setBody(new
                                RpcBlockBody(block)).build();
                        packetSender.sendGroup(blockPacket);
                        return;
                    } else {
                        log.error("get request failed, key={}\n", key);
                    }
                } catch (Exception ex) {
                    log.error("getResponse error", ex);
                }
            }
        }
    }

    @Override
    public BlockPacket get(String key) {
        ServerMessage.GetRequest getRequest = ServerMessage.GetRequest.newBuilder()
                .setKey(key).build();
        ServerMessage.GetResponse getResponse = RaftServerServiceContainer.getInstance().get(getRequest);
        return Json.toBean(getResponse.getValue(), BlockPacket.class);
    }
}
